#include "config.h"

#include <sys/types.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <stdarg.h>

#define DBG_SUBSYS S_LIBSTORAGE

#include "sysy_lib.h"
#include "cache.h"
#include "get_version.h"
#include "cluster.h"
#include "storage.h"
//#include "metadata.h"
#include "core.h"
#include "lich_md.h"
#include "volume.h"
#include "configure.h"
#include "job_dock.h"
#include "lichstor.h"
#include "chunk.h"
#include "ynet_rpc.h"
#include "vnode.h"
#include "net_global.h"
#include "bh.h"
#include "ylog.h"
#include "dbg.h"

/*
struct of file
  ------------------------------------------------------------------
  | file head (first chunk) | chunk | chunk |
  ------------------------------------------------------------------
*/

typedef vnode_entry_t entry_t;

static mcache_t *__vnode_cache__ = NULL;

static inline uint32_t __mode(uint32_t mode)
{
        if (mode > MODE_MAX) {
                DWARN("got mode %o\n", mode);
        }

        return mode & MODE_MAX;
}

static int __entry_cmp(const void *s1, const void *s2)
{
        const fileid_t *id = s1;
        const entry_t *ent = s2;

        return !oid_cmp(id, &ent->id);
}

static uint32_t __hash(const void *key)
{
        return ((fileid_t *)key)->id;
}

static uint32_t __core_hash(const void *key)
{
        const chkid_t *id = key;

        return core_hash(id);
}

static int __entry_init(entry_t **_ent, const char *pool, const fileinfo_t *fileinfo,
                        const nid_t *master, time_t ltime, const fileid_t *oid)
{
        int ret;
        entry_t *ent;

        ret = ymalloc((void **)&ent, sizeof(entry_t));
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ent->fileinfo = *fileinfo;
        ent->ltime = ltime;
        ent->master = *master;
        ent->id = *oid;
        ent->info_version = 0;
        strcpy(ent->pool, pool);

        *_ent = ent;

        return 0;
err_ret:
        return ret;
}

static void __entry_free(entry_t *ent)
{
        yfree((void **)&ent);
}

static int __drop(void *value, mcache_entry_t *cent, int recycle)
{
        entry_t *ent;

        (void) cent;
        ent = value;

        if (ent) {
                if (recycle) {
                        entry_t *entry = value;
                        DINFO("recycle "CHKID_FORMAT"\n", &entry->id);
                }
                __entry_free(ent);
        }

        return 0;
}

void vnode_release(mcache_entry_t *cent)
{
        mcache_release(cent);
}

STATIC int __vnode_add(const char *pool, const fileinfo_t *fileinfo, const nid_t *master, time_t ltime,
                        const fileid_t *oid, uint64_t info_version)
{
        int ret;
        entry_t *ent;

        DBUG("file "CHKID_FORMAT" len %llu\n", CHKID_ARG(oid), (LLU)fileinfo->size);

        ret = __entry_init(&ent, pool, fileinfo, master, ltime, oid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ent->info_version = info_version;
        ent->last_update = gettime();

        ret = mcache_insert(__vnode_cache__, oid, ent);
        if (unlikely(ret)) {
                __entry_free(ent);
                if (ret == EEXIST) {
                        goto out;
                } else
                        GOTO(err_ret, ret);
        }

out:
        return 0;
err_ret:
        return ret;
}

STATIC int __vnode_load_info(const fileid_t *id, fileinfo_t *fileinfo)
{
        int ret;

        ret = md_getattr(id, fileinfo);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

STATIC int __vnode_register(const char *pool, fileinfo_t *fileinfo, time_t *ltime, nid_t *master,
                            const fileid_t *id, uint64_t *info_version)
{
        int ret;
        nid_t *nid;
        char buf[CHKINFO_MAX];
        char tmp[MAX_NAME_LEN];
        chkinfo_t *chkinfo;

        if (unlikely(id->id == OID_NULL)) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        /*XXX: need cache*/
        /*get location for file head*/
        chkinfo = (void *)buf;
        ret = md_chunk_getinfo(pool, NULL, id, chkinfo, NULL);//(&nid, id, info_version);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        nid = &chkinfo->diskid[0].id;
        *info_version = chkinfo->info_version;

        ret = __vnode_load_info(id, fileinfo);
        if (unlikely(ret)) {
                if (id->type == __VOLUME_CHUNK__) {
                        sprintf(tmp, OBJECT_OFFLINE""CHKID_FORMAT"/", CHKID_ARG(id));
                        //_set_text(tmp, NULL, 0, O_CREAT);
                        path_validate(tmp, 1, 1);
                }

                GOTO(err_ret, ret);
        }

        ret = network_ltime(nid, ltime);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        *master = *nid;

        return 0;
err_ret:
        return ret;
}

STATIC int __vnode_lease_check_nolock(entry_t *ent)
{
        int ret;
        fileinfo_t *fileinfo;
        time_t ltime;
        char buf[MAX_BUF_LEN];

        ltime = 0;
        (void) network_ltime(&ent->master, &ltime);

        /* Cache must be reload from fileinfos in two cases:
         *     1. The connection to fileinfos had closed.
         *     2. Cache is notified to pull data from fileinfos every time.
         */
        if (ent->ltime == 0 || ltime != ent->ltime || gettime() -  ent->last_update > 10) {
                if (ent->ltime == 0 || ltime != ent->ltime) {
                        DINFO("reload file "CHKID_FORMAT", ltime %d, ent->ltime %d\n",
                              CHKID_ARG(&ent->id), (int)ltime, (int)ent->ltime);
                }

                fileinfo = (void *)buf;
                ret = __vnode_register(ent->pool, fileinfo, &ent->ltime, &ent->master,
                                       &ent->id, &ent->info_version);
                if (unlikely(ret))
                        GOTO(err_ret, ret);

                ent->fileinfo = *fileinfo;
                ent->last_update = gettime();
        }

        return 0;
err_ret:
        return ret;
}

STATIC int __vnode_lease_need_check(mcache_entry_t *cent, int *needcheck)
{
        int ret;
        time_t ltime;
        entry_t *ent;

        ret = mcache_rdlock(cent);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ent = cent->value;

        ltime = 0;
        (void) network_ltime(&ent->master, &ltime);

        if (ent->ltime == 0 || ltime != ent->ltime || gettime() -  ent->last_update > 10) {
                *needcheck = 1;
        } else
                *needcheck = 0;

        mcache_unlock(cent);

        return 0;
err_ret:
        return ret;
}

int vnode_force_reload_nolock(entry_t *ent)
{
        int ret;
        fileinfo_t *fileinfo;
        char buf[BUF_LEN];

        DINFO("force reload file "CHKID_FORMAT", ent->ltime %d\n",
              CHKID_ARG(&ent->id), (int)ent->ltime);

        fileinfo = (void *)buf;
        ret = __vnode_register(ent->pool, fileinfo, &ent->ltime, &ent->master, &ent->id,
                               &ent->info_version);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ent->fileinfo = *fileinfo;

        DBUG("force reload file "CHKID_FORMAT", ent->ltime %d\n",
             CHKID_ARG(&ent->id), (int)ent->ltime);

        return 0;
err_ret:
        return ret;
}

STATIC int __vnode_lease_check(mcache_entry_t *cent)
{
        int ret, needcheck;

        ret = __vnode_lease_need_check(cent, &needcheck);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (unlikely(needcheck)) {
                ret = mcache_wrlock(cent);
                if (unlikely(ret))
                        GOTO(err_ret, ret);

                ret = __vnode_lease_check_nolock(cent->value);
                if (unlikely(ret)) {
                        mcache_drop_nolock(cent);
                        GOTO(err_lock, ret);
                }

                mcache_unlock(cent);
        }

        return 0;
err_lock:
        mcache_unlock(cent);
err_ret:
        return ret;
}

int vnode_get(mcache_entry_t **_cent, const char *pool, const fileid_t *oid, int _retry)
{
        int ret, retry = 0;
        mcache_entry_t *cent;
        fileinfo_t *fileinfo;
        char buf[MAX_BUF_LEN];
        time_t ltime;
        nid_t master;
        uint64_t info_version;

        if (unlikely(oid->id == OID_NULL)) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

retry:
        ret = mcache_get(__vnode_cache__, oid, &cent);
        if (unlikely(ret)) {
                if (ret == ENOENT) {
                        fileinfo = (void *)buf;

                        ret = __vnode_register(pool, fileinfo, &ltime, &master, oid, &info_version);
                        if (unlikely(ret)) {
                                if (ret == EAGAIN) {
                                        if (_retry && !ng.daemon) {
                                                USLEEP_RETRY(err_ret, ret, retry, retry, 30, 1000);
                                        } else
                                                GOTO(err_ret, ret);
                                } else
                                        GOTO(err_ret, ret);
                        }

                        ret = __vnode_add(pool, fileinfo, &master, ltime, oid, info_version);
                        if (unlikely(ret)) {
                                if (ret == EEXIST) {
                                        DWARN("file "CHKID_FORMAT" exist\n", CHKID_ARG(oid));
                                        goto retry;
                                } else
                                        GOTO(err_ret, ret);
                        }

                        ret = mcache_get(__vnode_cache__, oid, &cent);
                        if (unlikely(ret))
                                GOTO(err_ret, ret);

                        DBUG("load "CHKID_FORMAT" size %llu\n",
                             CHKID_ARG(oid), (LLU)fileinfo->size);
                } else
                        GOTO(err_ret, ret);
        } else {
                ret = __vnode_lease_check(cent);
                if (unlikely(ret)) {
                        GOTO(err_release, ret);
                }
        }

        *_cent = cent;

        return 0;
err_release:
        vnode_release(cent);
err_ret:
        return ret;
}

int vnode_getattr(const char *pool, const fileid_t *id, fileinfo_t *fileinfo, int retry)
{
        int ret;
        mcache_entry_t *cent;
        entry_t *ent;

        DBUG("id "CHKID_FORMAT"\n", CHKID_ARG(id));

        ret = vnode_get(&cent, pool, id, retry);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = mcache_rdlock(cent);
        if (unlikely(ret))
                GOTO(err_release, ret);

        ent = cent->value;
        *fileinfo = ent->fileinfo;

        mcache_unlock(cent);
        vnode_release(cent);

        return 0;
err_release:
        vnode_release(cent);
err_ret:
        return _errno(ret);
}

int vnode_location(const char *pool, const fileid_t *id, nid_t *nid)
{
        int ret;
        mcache_entry_t *cent;
        entry_t *ent;

        if (id->type != __VOLUME_CHUNK__)
                DINFO("id "CHKID_FORMAT"\n", CHKID_ARG(id));

        ret = vnode_get(&cent, pool, id, 1);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = mcache_rdlock(cent);
        if (unlikely(ret))
                GOTO(err_release, ret);

        ent = cent->value;
        *nid = ent->master;

        mcache_unlock(cent);
        vnode_release(cent);

        return 0;
err_release:
        vnode_release(cent);
err_ret:
        return _errno(ret);
}

int vnode_init()
{
        int ret;

        ret = mcache_init(&__vnode_cache__, MCACHE_SIZE, __entry_cmp, __hash, __core_hash,
                          __drop, 0, "vnode");
        if (unlikely(ret))
                GOTO(err_ret, ret);

        YASSERT(__vnode_cache__);

        return 0;
err_ret:
        return ret;
}

int vnode_drop(mcache_entry_t *cent, const fileid_t *id)
{
        int ret;

        ret = mcache_lockcache(__vnode_cache__, id);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        mcache_drop_nolock(cent);

        mcache_unlockcache(__vnode_cache__, id);

        return 0;
err_ret:
        return ret;
}

int vnode_reload(const char *pool, const fileid_t *oid)
{
        int ret;
        mcache_entry_t *cent;
        entry_t *ent;

        ret = vnode_get(&cent, pool, oid, 1);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = mcache_wrlock(cent);
        if (unlikely(ret))
                GOTO(err_release, ret);

        ent = cent->value;

        ret = vnode_force_reload_nolock(ent);
        if (unlikely(ret)) {
                mcache_drop_nolock(cent);
                GOTO(err_lock, ret);
        }

        mcache_unlock(cent);
        vnode_release(cent);

        return 0;
err_lock:
        mcache_unlock(cent);
err_release:
        vnode_release(cent);
err_ret:
        return _errno(ret);
}

STATIC int __vnode_chown(entry_t *ent, int uid, int gid)
{
        int ret;
        setattr_t setattr;
        fileinfo_t fileinfo;

        memset(&setattr, 0x0, sizeof(setattr));
        setattr.uid.set_it = 1;
        setattr.gid.set_it = 1;
        setattr.uid.val = uid;
        setattr.gid.val = gid;

        ret = md_setattr(&ent->id, &fileinfo, &setattr);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ent->fileinfo = fileinfo;

        return 0;
err_ret:
        return ret;
}

int vnode_chown(const char *pool, const fileid_t *id, int uid, int gid)
{
        int ret, retry = 0;
        entry_t *ent;
        mcache_entry_t *cent;

        ret = vnode_get(&cent, pool, id, 1);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = mcache_wrlock(cent);
        if (unlikely(ret))
                GOTO(err_release, ret);

        ent = cent->value;

retry:
        ret = __vnode_chown(ent, uid, gid);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_lock, ret, retry, retry, 10, (100 * 1000));
                } else
                        GOTO(err_lock, ret);
        }

        mcache_unlock(cent);
        vnode_release(cent);

        return 0;
err_lock:
        mcache_unlock(cent);
err_release:
        vnode_release(cent);
err_ret:
        return ret;
}

STATIC int __vnode_chmod(entry_t *ent, int mode)
{
        int ret;
        setattr_t setattr;
        fileinfo_t fileinfo;

        memset(&setattr, 0x0, sizeof(setattr));
        setattr.mode.set_it = 1;
        setattr.mode.val = mode;

        ret = md_setattr(&ent->id, &fileinfo, &setattr);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ent->fileinfo = fileinfo;

        return 0;
err_ret:
        return ret;
}

int vnode_chmod(const char *pool, const fileid_t *id, int mode)
{
        int ret, retry = 0;
        entry_t *ent;
        mcache_entry_t *cent;

        ret = vnode_get(&cent, pool, id, 1);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = mcache_wrlock(cent);
        if (unlikely(ret))
                GOTO(err_release, ret);

        ent = cent->value;

retry:
        ret = __vnode_chmod(ent, mode);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_lock, ret, retry, retry, 10, (100 * 1000));
                } else
                        GOTO(err_lock, ret);
        }

        mcache_unlock(cent);
        vnode_release(cent);

        return 0;
err_lock:
        mcache_unlock(cent);
err_release:
        vnode_release(cent);
err_ret:
        return ret;
}

STATIC int __vnode_utime(entry_t *ent, int atime, int mtime)
{
        int ret;
        setattr_t setattr;
        fileinfo_t fileinfo;

        memset(&setattr, 0x0, sizeof(setattr));
        setattr.atime.set_it = 1;
        setattr.mtime.set_it = 1;
        setattr.atime.time.seconds = atime;
        setattr.mtime.time.seconds = mtime;

        ret = md_setattr(&ent->id, &fileinfo, &setattr);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ent->fileinfo = fileinfo;

        return 0;
err_ret:
        return ret;
}

int vnode_utime(const char *pool, const fileid_t *id, int atime, int mtime)
{
        int ret, retry = 0;
        entry_t *ent;
        mcache_entry_t *cent;

        ret = vnode_get(&cent, pool, id, 1);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = mcache_wrlock(cent);
        if (unlikely(ret))
                GOTO(err_release, ret);

        ent = cent->value;

retry:
        ret = __vnode_utime(ent, atime, mtime);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_lock, ret, retry, retry, 10, (100 * 1000));
                } else
                        GOTO(err_lock, ret);
        }

        mcache_unlock(cent);
        vnode_release(cent);

        return 0;
err_lock:
        mcache_unlock(cent);
err_release:
        vnode_release(cent);
err_ret:
        return ret;
}

int vnode_isempty(const fileid_t *fid, int *empty)
{
        int ret;
        filestat_t filestat;

        ret = md_stat(fid, &filestat);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        *empty = (filestat.sparse == filestat.chknum);

        return 0;
err_ret:
        return ret;
}

STATIC int __vnode_set_repnum(entry_t *ent, int repnum)
{
        int ret;
        setattr_t setattr;
        fileinfo_t fileinfo;

        memset(&setattr, 0x0, sizeof(setattr));
        setattr.replica.set_it = 1;
        setattr.replica.val = repnum;

        ret = md_setattr(&ent->id, &fileinfo, &setattr);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ent->fileinfo = fileinfo;

        return 0;
err_ret:
        return ret;
}

int vnode_set_repnum(const char *pool, const fileid_t *id, int repnum)
{
        int ret;
        mcache_entry_t *cent;

        //if (repnum < LICH_REPLICA_MIN || repnum > LICH_REPLICA_MAX) {
        if (repnum < 1 || repnum > LICH_REPLICA_MAX) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        ret = vnode_get(&cent, pool, id, 1);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = mcache_wrlock(cent);
        if (unlikely(ret))
                GOTO(err_release, ret);

        ret = __vnode_set_repnum(cent->value, repnum);
        if (unlikely(ret)) {
                GOTO(err_lock, ret);
        }

        mcache_unlock(cent);
        vnode_release(cent);

        return 0;
err_lock:
        mcache_unlock(cent);
err_release:
        vnode_release(cent);
err_ret:
        return ret;
}

int vnode_setattr(const char *pool, const fileid_t *id, const setattr_t *setattr)
{
        int ret;
        mcache_entry_t *cent;
        entry_t *ent;
        fileinfo_t fileinfo;

        ret = vnode_get(&cent, pool, id, 1);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = mcache_wrlock(cent);
        if (unlikely(ret))
                GOTO(err_release, ret);

        ent = cent->value;
        ret = md_setattr(&ent->id, &fileinfo, setattr);
        if (unlikely(ret))
                GOTO(err_lock, ret);

        ent->fileinfo = fileinfo;

        mcache_unlock(cent);
        vnode_release(cent);

        return 0;
err_lock:
        mcache_unlock(cent);
err_release:
        vnode_release(cent);
err_ret:
        return ret;
}

int vnode_update(const char *pool, const fileid_t *id, uint64_t size, int force)
{
        int ret;
        mcache_entry_t *cent;
        entry_t *ent;

        ret = vnode_get(&cent, pool, id, 1);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = mcache_wrlock(cent);
        if (unlikely(ret))
                GOTO(err_release, ret);

        ent = cent->value;

        if (size > ent->fileinfo.size || force) {
                ent->fileinfo.size = size;
        }

        mcache_unlock(cent);
        vnode_release(cent);

        return 0;
err_release:
        vnode_release(cent);
err_ret:
        return ret;
}

int vnode_reset(const char *pool, const fileid_t *id)
{
        int ret;
        mcache_entry_t *cent;
        entry_t *ent;

        (void) pool;
        ret = mcache_get(__vnode_cache__, id, &cent);
        if (unlikely(ret)) {
                if (ret == ENOENT)
                        goto out;
                else
                        GOTO(err_ret, ret);
        }

        ret = mcache_wrlock(cent);
        if (unlikely(ret))
                GOTO(err_release, ret);

        ent = cent->value;
        ent->ltime = 0;

        mcache_unlock(cent);
        mcache_release(cent);

out:
        return 0;
err_release:
        mcache_release(cent);
err_ret:
        return ret;
}
